package com.ikun.search.listener;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.ikun.common.entity.Docker;
import com.ikun.search.dao.ElasticsearchDao;
import com.ikun.common.entity.ElasticsearchEntity;
import com.ikun.search.service.DockerService;
import com.ikun.search.service.impl.DockerServiceImpl;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;

/**
 * 客户端测试
 *
 * @author Guio
 */
@Component
public class DockerDatasourceListener {

    @Value("${canal.hostname}")
    private String hostname;
    @Value("${canal.port}")
    private Integer port;

    @Resource
    private DockerService dockerService;
    @Resource
    private ElasticsearchDao dao;

    //创建线程执行
    @EventListener(ApplicationStartedEvent.class)
    public void startDockerDataSourceListenerThread() {
        System.out.println("===> 程序启动，创建canal监听线程");
        Thread thread = new Thread(() -> {
            try {
                dockerDataSource();
            } catch (IOException e) {
                System.out.println("===> canal监听线程异常");
                e.printStackTrace();
            }
        });
        thread.start();
    }

    public void dockerDataSource() throws IOException {
        // 创建canal连接对象
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(hostname,
                port), "example", "canal", "canal");
        try {
            //连接
            connector.connect();
            //监听hos_docker库
            connector.subscribe("hos_docker\\..*");
//            connector.subscribe("hos_docker.docker");
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(1000);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    //没有数据，就休眠1秒
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    //有数据就打印
                    printEntry(message.getEntries());
                }
                // 提交确认
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<Entry> entrys) throws IOException {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN ||
                    entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR parse data:" + entry.toString(), e);
            }
            EventType eventType = rowChage.getEventType();  //数据库操作类型
            System.out.print("===>");
            System.out.printf("数据库binlog[%s:%s]",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset());
            System.out.printf("=== 数据库：%s；表：%s",
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
            System.out.printf("=== 操作类型eventType：%s", eventType);
            System.out.println("<===");
            String tableName = entry.getHeader().getTableName();
            //遍历行数据
            for (RowData rowData : rowChage.getRowDatasList()) {
                switch (tableName) {
                    case "docker":
                        //判断增删改操作
                        if (eventType == EventType.DELETE) {
                            //删除es索引
                            deleteDockerIndex(rowData.getBeforeColumnsList());
                        } else if (eventType == EventType.INSERT) {
                            updateDocker(rowData.getAfterColumnsList());
                        } else {
                            //更新es索引
                            updateDocker(rowData.getBeforeColumnsList());
                        }
                        break;
                    case "docker_disease":
                        //判断增删改操作
                        if (eventType == EventType.DELETE) {
                            //删除es索引
                            updateDockerDisease(rowData.getBeforeColumnsList());
                        } else if (eventType == EventType.INSERT) {
                            updateDockerDisease(rowData.getAfterColumnsList());
                        } else {
                            //更新es索引
                            updateDockerDisease(rowData.getBeforeColumnsList());
                        }
                        break;
                    case "disease":
                        System.out.println("===> 疾病表更新，初始化索引");
                        dockerService.initDockerIndex();
                        System.out.println("===> 疾病表更新初始化索引完成");
                        break;
                }
            }
        }
    }

    /**
     * 医生表修改更新
     */
    private void updateDocker(List<Column> columns) throws IOException {
        for (Column column : columns) {
            if ("id".equals(column.getName())) {
                updateIndex(column.getValue());
            }
        }
    }

    /**
     * 医生疾病关系表修改更新
     */
    private void updateDockerDisease(List<Column> columns) throws IOException {
        for (Column column : columns) {
            if ("docker_id".equals(column.getName())) {
                updateIndex(column.getValue());
            }
        }
    }

    /**
     * 更新ES数据请求方法
     */
    private void updateIndex(String dockerId) throws IOException {
        System.out.println("数据更新：===> 医生id:" + dockerId);
        Docker docker = dockerService.getDockerById(Integer.valueOf(dockerId));
        ElasticsearchEntity dockerEntity = new ElasticsearchEntity(dockerId, docker);
        dao.saveOrUpdate(DockerServiceImpl.INDEX_NAME, dockerEntity);
    }

    /**
     * 删除ES数据
     */
    private void deleteDockerIndex(List<Column> columns) throws IOException {
        for (Column column : columns) {
            if ("id".equals(column.getName())) {
                String id = column.getValue();
                System.out.println("数据删除：===> 影响id:" + id);
                dao.deleteById(DockerServiceImpl.INDEX_NAME, id);
            }
        }
    }

}